package net.bwie.realtime.jtp.dwd.log.functionlx;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class AppLog extends ProcessFunction<String, String> {

    private final OutputTag<String> errorTag;
    private final OutputTag<String> startTag;
    private final OutputTag<String> displayTag;
    private final OutputTag<String> actionTag;

    public AppLog(OutputTag<String> errorTag, OutputTag<String> startTag, OutputTag<String> displayTag, OutputTag<String> actionTag) {
        this.errorTag = errorTag;
        this.startTag = startTag;
        this.displayTag = displayTag;
        this.actionTag = actionTag;
    }


    @Override
    public void processElement(String value,Context ctx, Collector<String> out) throws Exception {
        // 第一步：解析JSON数据
        JSONObject jsonObject = JSON.parseObject(value);

        // 第2步：错误日志
        if(null != jsonObject.getString("err")){
            // 侧边流输出
            ctx.output(errorTag,value);
            // 错误数据已经单独保存,无需在保存,直接删除
            jsonObject.remove("err");
        }

        // 第三补 启动日志
        if(null != jsonObject.getString("start")){
            // 侧边流输出
            ctx.output(startTag, jsonObject.toJSONString());
        }
        // 第4步：页面日志
        else{
            // 获取曝光日志和动作日志
            JSONArray display = jsonObject.getJSONArray("displays");
            JSONArray action = jsonObject.getJSONArray("actions");

            // 删除曝光和动作
            jsonObject.remove("displays");
            jsonObject.remove("actions");

            // 输出页面日志
            out.collect(jsonObject.toJSONString());

            // 第5步：曝光日志
            if(null != display && !display.isEmpty()){
                for(int i=0;i < display.size(); i++){
                    JSONObject jsonObject1 = display.getJSONObject(i);
                    // 添加字段
                    jsonObject1.put("displays",display);
                    // 输出
                    ctx.output(displayTag,jsonObject.toJSONString());
                }
            }


            // 删除曝光
            jsonObject.remove("display");

            // 第六步:动作日志
            if(null != action && !action.isEmpty()){
                for(int i=0;i<action.size();i++){
                    JSONObject jsonObject1 = action.getJSONObject(i);
                    // 添加字段
                    jsonObject1.put("action",action);
                    // 输出动作日志
                    ctx.output(actionTag,jsonObject.toJSONString());
                }
            }
        }
    }
}
